package com.hitao.bolts;

import org.apache.storm.state.State;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.*;
import org.apache.storm.tuple.Tuple;

import java.util.Map;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;

/**
 * Created by Administrator on 2016/5/3.
 */
public class BatcherBolt implements IRichBolt {

    private static final long serialVersionUID = 1L;
    private OutputCollector collector;
    private Queue<Tuple> tupleQueue = new ConcurrentLinkedQueue<Tuple>();
    private int count;
    private int timeEclipse;
    private long lastTime;

    public BatcherBolt(int i, int eclipseInSecondes) {
        count = i;
        timeEclipse = eclipseInSecondes;
        lastTime = System.currentTimeMillis();
    }

    @Override
    public void prepare(Map map, TopologyContext topologyContext, OutputCollector outputCollector) {
        this.collector = collector;
    }

    @Override
    public void execute(Tuple tuple) {
        tupleQueue.add(tuple);
        long currentTime = System.currentTimeMillis();
        if (tupleQueue.size() > count || currentTime - lastTime > timeEclipse*1000) {
            System.out.println("Do batching Opertions!!!");
            for (int i = 0; i < count; i++) {
                Tuple tup = (Tuple) tupleQueue.poll();
                System.out.println("process:" + tup);
            }
            lastTime = currentTime;
        } else {
            System.out.println("Cached Tuple:"+tuple+"curSze:"+tupleQueue.size());
        }
    }

    @Override
    public void cleanup() {
        System.out.println("End===========================");
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {

    }

    @Override
    public Map<String, Object> getComponentConfiguration() {
        return null;
    }
}
